package com.gis.bigdata.spark02.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author LnnuUser
 * @create 2022-10-05-下午10:11
 */
object Scala_Test01_1 {

  /*
  日期，用户id，session ID，页面id，动作时间，搜索关键字，点击品类id，产品id，下单品类id、支付品类id，产品id，城市id
  2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19
  2019-07-17,95,26070e87-1ad7-49a3-8fb3-cc741facaddf,48,2019-07-17 00:00:10,null,16,98,null,null,null,null,19
   */
  def main(args: Array[String]): Unit = {
    // 需求1：Top10热门品类
    // TODO: TOP10
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
    val sc: SparkContext = new SparkContext(sparkConf)

    // TODO 1. 读取原始数据
    val actionRDD: RDD[String] = sc.textFile("/home/lnnu/IdeaProjects/spark/datas/user_visit_action.txt")
    // Q1. RDD重复使用
    // TODO Q1. 缓存
    actionRDD.cache()

    // TODO 2. 统计品类的点击数量：（品类id，点击数量）
    val clickActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        datas(6) != "-1"
      }
    )

    val clickCountRDD: RDD[(String, Int)] = clickActionRDD.map(
      action => {
        val datas: Array[String] = action.split("_")
        (datas(6), 1)
      }
    ).reduceByKey(_ + _)

    // TODO 3. 统计品类的下单数量：（品类id，下单数量）
    val orderActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        datas(8) != "null"
      }
    )

    // orderid => 1,2,3
    // -> [(1,1),(2,1),(3,1)]
    val orderCountRDD: RDD[(String, Int)] = orderActionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        val cid: String = datas(8)
        val cids: Array[String] = cid.split(",")
        cids.map(id => (id, 1))
      }
    ).reduceByKey(_ + _)

    // TODO 4. 统计品类的支付数量：（品类id，支付数量）
    val payActionRDD: RDD[String] = actionRDD.filter(
      action => {
        val datas: Array[String] = action.split("_")
        datas(10) != "null"
      }
    )

    // orderid => 1,2,3
    // -> [(1,1),(2,1),(3,1)]
    val payCountRDD: RDD[(String, Int)] = payActionRDD.flatMap(
      action => {
        val datas: Array[String] = action.split("_")
        val cid: String = datas(10)
        val cids: Array[String] = cid.split(",")
        cids.map(id => (id, 1))
      }
    ).reduceByKey(_ + _)

    // TODO 5. 将品类进行排序，并且取前10
    // 点击数量排序、下单数量排序、支付数量的排序
    // 元组的排序：先比较第一个，在比较第二个，在比较第三个，以此类推
    // （（品类id），（点击数量，下单数量，支付数量））
    // join:两个不同的数据源连接在一起，这里不可以使用，类似于内连接，会缺少数据
    // zip:拉链，不可以用
    // leftOuterJoin：不可以用
    // cogroup：connect+group，可以用，数据不存在，id也会存在
    val rdd1: RDD[(String, (Int, Int, Int))] = clickCountRDD.map {
      case (cid, cnt) => {
        (cid, (cnt, 0, 0))
      }
    }

    val rdd2: RDD[(String, (Int, Int, Int))] = orderCountRDD.map {
      case (oid, cnt) => {
        (oid, (0, cnt, 0))
      }
    }

    val rdd3: RDD[(String, (Int, Int, Int))] = payCountRDD.map {
      case (pid, cnt) => {
        (pid, (0, 0, cnt))
      }
    }

    // 将三个数据源合并在一起
    val soruceRDD: RDD[(String, (Int, Int, Int))] = rdd1.union(rdd2).union(rdd3)

    val analysisRDD: RDD[(String, (Int, Int, Int))] = soruceRDD.reduceByKey(
      (t1, t2) => {
        (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
      }
    )

    // TODO 6. 将结果采集到控制台打印
    val resultRDD: Array[(String, (Int, Int, Int))] = analysisRDD.sortBy(_._2, false).take(10)

    resultRDD.foreach(println)

    sc.stop()
  }
}
